package com.pangyangyang.elasticsearch.plugin.canal.collector;

import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
import com.pangyangyang.elasticsearch.plugin.util.MessageRiverConstants;
import com.pangyangyang.elasticsearch.plugin.collector.AbstractEventCollector;
import com.pangyangyang.elasticsearch.plugin.collector.EventCollector;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;

import java.net.InetSocketAddress;

/**
 * Created by yangwang on 2015/1/15.
 */
public class CanalCollector extends AbstractEventCollector implements EventCollector {

    private  SimpleCanalConnector connector;
    private CanalEventTask canalEventTask;
    private String host;
    private int port;
    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    public CanalCollector(MessageRiverSettings settings) {
        super(settings);
        host = settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_HOST);
        port = settings.getInt(MessageRiverConstants.SETTING_KEY_COMMON_PORT, 11111);
    }

    @Override
    public void start() {
        String username = settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_USERNAME);
        String password = settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_PASSWORD);
        String destination = settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_DESTINATION);

        logger.debug("canal collector trying to connect.");
        connector = new SimpleCanalConnector(new InetSocketAddress(host,
                port), username, password, destination);
        connector.setSoTimeout(30 * 1000);
        connector.connect();
        connector.subscribe();
        this.canalEventTask = new CanalEventTask(connector, settings);
        Thread subscribeThread = new Thread(this.canalEventTask);
        subscribeThread.setName("Canal-subscribe-TH");
        subscribeThread.start();

    }

    @Override
    public void stop() {
        logger.info("stopping canal collector {}-{}.",
                settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_HOST), settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_PORT));
        canalEventTask.stop();
        logger.info("stopped canal collector {}-{}.",
                settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_HOST), settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_PORT));
        connector.disconnect();
        logger.info("disconnected canal {}-{}.",
                settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_HOST), settings.getString(MessageRiverConstants.SETTING_KEY_COMMON_PORT));
    }

    @Override
    public String getResourceDigest() {
        return host + MessageRiverConstants.COMMON_SPLITTER + port;
    }

}
